package com.unisound.tracelog.client.core.listener;

import com.unisound.tracelog.client.core.properties.TraceLogClientProperties;
import com.unisound.tracelog.common.model.span.Span;
import com.unisound.tracelog.common.model.span.Tag;
import com.unisound.tracelog.common.utils.JsonUtils;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.List;

public class KafkaTraceLogListener implements TraceLogListener {
    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaTraceLogListener.class);

    private KafkaTemplate<String, String> kafkaTemplate;
    private TraceLogClientProperties traceLogClientProperties;

    public KafkaTraceLogListener(KafkaTemplate<String, String> kafkaTemplate, TraceLogClientProperties traceLogClientProperties) {
        this.kafkaTemplate = kafkaTemplate;
        this.traceLogClientProperties = traceLogClientProperties;
    }

    public void listenerSpanEvent(SpanEvent spanEvent) {
        if (!traceLogClientProperties.isReportEnable()) return;

        if (spanEvent == null) return;
        Span span = spanEvent.getSpan();
        if (span == null) return;

        String data = JsonUtils.toJson(span);
        if (data == null) return;

        send(traceLogClientProperties.getKafkaTopicSpan(), span.getTraceId().toString(), data);
    }

    @Override
    public void listenerTagEvent(TagEvent tagEvent) {
        if (!traceLogClientProperties.isReportEnable()) return;
        if (!checkTagEvent(tagEvent)) return;

        List<Tag> tags = tagEvent.getTags();
        String data = JsonUtils.toJson(tags);
        if (StringUtils.isEmpty(data)) return;

        send(traceLogClientProperties.getKafkaTopicTag(), tags.get(0).getTraceId().toString(), data);
    }

    private boolean checkTagEvent(TagEvent tagEvent) {
        if (tagEvent == null) return false;
        if (CollectionUtils.isEmpty(tagEvent.getTags())) return false;
        return true;
    }

    private void send(String topic, String key, String data) {
        try {
            kafkaTemplate.send(topic, key, data);
        } catch (Exception e) {
            LOGGER.error("TraceLog异步发送到kafka失败. topic={}, key={}, data={}", topic, key, data, e);
        }
    }

}
